# @Version        : 1.0
# @Update Time    : 2025/4/3 23:45
# @File           : redis_utils.py
# @IDE            : PyCharm
# @Desc           : 文件描述信息
import sys
from typing import Any, Dict, Optional, Tuple, Union
from redis import asyncio as aioredis
import redis
from redis.connection import ConnectionPool
from redis import Redis, RedisCluster
from redis.asyncio import Redis as AsyncRedis, RedisCluster as AsyncRedisCluster
from redis.exceptions import RedisError, AuthenticationError, TimeoutError
from src.log import logger

# 类型别名
RedisClient = Union[Redis, RedisCluster]
AsyncRedisClient = Union[AsyncRedis, AsyncRedisCluster]
ConnectionKey = Tuple[Any, ...]


class BaseRedisManager:
    """Redis 连接管理基类"""

    _pool_dict: Dict[ConnectionKey, ConnectionPool] = {}

    def __init__(
        self,
        host: str = "127.0.0.1",
        port: int = 6379,
        db: int = 0,
        password: Optional[str] = None,
        decode_responses: bool = True,
        max_connections: int = 50,
        **extra: Any,
    ):
        """初始化连接参数

        :param host: Redis 地址
        :param port: Redis 端口
        :param db: 数据库编号
        :param password: 认证密码
        :param decode_responses: 自动解码响应
        :param max_connections: 最大连接数
        :param extra: 额外连接参数
        """
        self.conn_key = (host, port, db, password, decode_responses, *extra.items())
        self.host = host
        self.port = port
        self.db = db
        self.password = password
        self.decode_responses = decode_responses
        self.max_connections = max_connections
        self.extra = extra


class SyncRedisManager(BaseRedisManager):
    """同步 Redis 连接管理器"""

    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self._client: Optional[RedisClient] = None
        self._init_pool()
        self._verify_connection()

    def _init_pool(self):
        """初始化连接池"""
        if self.conn_key not in self._pool_dict:
            self._pool_dict[self.conn_key] = redis.ConnectionPool(
                host=self.host,
                port=self.port,
                db=self.db,
                password=self.password,
                decode_responses=self.decode_responses,
                max_connections=self.max_connections,
                **self.extra,
            )

    def _verify_connection(self):
        """验证连接有效性"""
        try:
            self.client().ping()
            logger.info("Redis 连接成功")
        except AuthenticationError as e:
            logger.error(f"Redis 认证失败: {e}")
            sys.exit()
        except TimeoutError as e:
            logger.error(f"Redis 连接超时: {e}")
            sys.exit()
        except RedisError as e:
            logger.error(f"Redis connection failed: {e}")
            sys.exit()

    def client(self) -> RedisClient:
        """获取 Redis 客户端"""
        if not self._client:
            if "cluster" in self.extra:
                self._client = redis.RedisCluster(
                    host=self.host,
                    port=self.port,
                    password=self.password,
                    decode_responses=self.decode_responses,
                    **self.extra,
                )
            else:
                self._client = redis.Redis(
                    connection_pool=self._pool_dict[self.conn_key]
                )
        return self._client


class AsyncRedisManager(BaseRedisManager):
    """异步 Redis 连接管理器"""

    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self._client: Optional[AsyncRedisClient] = None
        self._init_pool()

    def _init_pool(self):
        """初始化异步连接池"""
        if self.conn_key not in self._pool_dict:
            self._pool_dict[self.conn_key] = aioredis.ConnectionPool(
                host=self.host,
                port=self.port,
                db=self.db,
                password=self.password,
                decode_responses=self.decode_responses,
                max_connections=self.max_connections,
                **self.extra,
            )

    async def _verify_connection(self):
        """异步验证连接"""
        try:
            client = await self.client()
            await client.ping()
        except ConnectionError as e:
            raise RedisError(f"Redis connection failed: {e}") from e

    async def client(self) -> AsyncRedisClient:
        """获取异步客户端"""
        if not self._client:
            if "cluster" in self.extra:
                self._client = await aioredis.RedisCluster(
                    host=self.host,
                    port=self.port,
                    password=self.password,
                    decode_responses=self.decode_responses,
                    **self.extra,
                )
            else:
                self._client = aioredis.Redis(
                    connection_pool=self._pool_dict[self.conn_key]
                )
            await self._verify_connection()
        return self._client


class RedisMixin:
    """同步 Redis 混入类"""

    def __init__(self, **redis_kwargs):
        self._sync_redis: Optional[SyncRedisManager] = None
        self.redis_config = redis_kwargs

    def redis(self) -> RedisClient:
        """获取同步 Redis 客户端"""
        if not self._sync_redis:
            self._sync_redis = SyncRedisManager(**self.redis_config)
        return self._sync_redis.client()


class AsyncRedisMixin:
    """异步 Redis 混入类"""

    def __init__(self, **redis_kwargs):
        self._async_redis: Optional[AsyncRedisManager] = None
        self.redis_config = redis_kwargs

    async def redis(self) -> AsyncRedisClient:
        """获取异步 Redis 客户端"""
        if not self._async_redis:
            self._async_redis = AsyncRedisManager(**self.redis_config)
        return await self._async_redis.client()


# 初始化异步Redis连接池
async_redis_pool = AsyncRedisMixin()
